/*
 * HttpMediaCasterConnection.cpp
 *
 *  Created on: 2019年3月1日
 *      Author: terry
 */

#include "HttpMediaCasterConnection.h"

#include "CLog.h"
#include "TStringUtil.h"
#include "Socket.h"
#include <assert.h>
#include <StopWatch.h>
#include "MediaCasterConfig.h"



namespace av
{


HttpMediaCasterConnection::HttpMediaCasterConnection():
    m_streamState(av::STATE_STOPPED),
    m_jamming(),
    m_maxCacheSize(),
    m_empty()
{
    m_maxCacheSize = 1024 * 1000;
}

HttpMediaCasterConnection::~HttpMediaCasterConnection()
{
}


int HttpMediaCasterConnection::onOpen()
{
    bufferevent* bev = getBufferEvent();
    bufferevent_setwatermark(bev, EV_WRITE, 1024 * 16, m_maxCacheSize);

    //bufferevent_enable(bev, EV_WRITE);

    evutil_socket_t fd = bufferevent_getfd(bev);
    evutil_make_socket_nonblocking(fd);

//    int value = true;
//    int ret = ::setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char*)&value, sizeof(value));

    comn::Socket sock;
    sock.attach(fd);
    if (MediaCasterConfig::SEND_BUFFER_SIZE > 0)
    {
        sock.setSendBufferSize(MediaCasterConfig::SEND_BUFFER_SIZE);
    }

    comn::SockAddr peerAddr = sock.getPeerName();
    m_peerID = comn::StringUtil::format("%s:%d", peerAddr.getIPAddr(), peerAddr.getPort());
    sock.detach();

    return 0;
}

void HttpMediaCasterConnection::onClose()
{
    CLog::debug("HttpMediaCasterConnection(%s) onClose stream%s\n",
        m_peerID.c_str(),
        m_streamName.c_str());

}

void HttpMediaCasterConnection::onWrite(struct bufferevent *bev)
{
    //
}

void HttpMediaCasterConnection::onRead(struct bufferevent *bev)
{
    size_t len = evbuffer_get_length(bufferevent_get_input(bev));

    m_recvBuffer.ensure(len);
    size_t count = bufferevent_read(bev, m_recvBuffer.data(), len);

    HttpMessageHandler handler = [&] (HttpMessage& msg, std::string& buf)
    {
        handleRecv(bev, msg, buf);
    };

    int ret = m_httpParser.parse((char*)m_recvBuffer.data(), count, handler);


    /*
    if (!m_headerReady)
    {
        if (len >= sizeof(MediaTransportHeader))
        {
            len = bufferevent_read(bev, &m_header, sizeof(MediaTransportHeader));
            assert(len == sizeof(MediaTransportHeader));
            m_headerReady = true;

            len = evbuffer_get_length(bufferevent_get_input(bev));
        }
    }

    if (!m_headerReady)
    {
        return;
    }

    size_t dataLen = m_header.length;
    if (len < dataLen)
    {
        return;
    }

    m_jsonBuffer.ensure(dataLen);
    bufferevent_read(bev, m_jsonBuffer.data(), dataLen);
    m_headerReady = false;

    std::string json(m_jsonBuffer.c_str(), dataLen);
    Json::Value req;
    Json::Reader reader;
    if (!reader.parse(json, req))
    {
        CLog::warning("TcpMediaCasterConnection(%s). invalid request.\n", m_peerID.c_str());
        close();
        return;
    }

    try
    {
        Json::Value resp;
        bool success = handleCommand(m_header, req, resp);

        {
            json = comn::StringCast::toString(resp);

            MediaTransport::setCommand(m_header, kMediaCmdResp);
            m_header.length = json.size();
            m_header.subtype = success ? kMediaCmdResp : kMediaCmdError;

            int ret = bufferevent_write(bev, (char*)&m_header, sizeof(m_header));
            bufferevent_write(bev, json.c_str(), json.size());
        }

        if (!success)
        {
            CLog::warning("TcpMediaCasterConnection(%s). bad command.\n", m_peerID.c_str());

            close();
        }
    }
    catch (std::exception& ex)
    {
        CLog::warning("TcpMediaCasterConnection(%s). execption %s\n", m_peerID.c_str(), ex.what());
        close();
    }
    */
}

void HttpMediaCasterConnection::onEvent(struct bufferevent *bev, short events)
{
    TcpConnection::onEvent(bev, events);
}


void HttpMediaCasterConnection::onMediaFormat(const av::MediaFormat& fmt)
{
    m_format = fmt;
}

void HttpMediaCasterConnection::onMediaPacket(av::MediaPacketPtr& pkt)
{
    comn::StopWatch watch;
    writeMedia(pkt);

    uint64_t elapse = watch.elapse();
    if (elapse > 20)
    {
        char pktType = pkt->isVideo() ? 'V' : 'A';
        CLog::warning("HttpMediaCasterConnection::onMediaPacket elapse too much. pkt type:%c, elapse:%d\n", pktType, elapse);
    }
}

void HttpMediaCasterConnection::onMediaEvent(int event)
{
    // pass
}

void HttpMediaCasterConnection::setStreamName(const std::string& name)
{
    m_streamName = name;
}

int HttpMediaCasterConnection::writeMedia(av::MediaPacketPtr& pkt)
{
    if (!canStream())
    {
        return 0;
    }

    bufferevent* bev = getBufferEvent();
    if (!bev)
    {
        return 0;
    }

    uint8_t* buf = pkt->data;
    int buf_size = pkt->size;

    evbuffer * evbuf = bufferevent_get_output(bev);
    size_t length = evbuffer_get_length(evbuf);

    if (length > m_maxCacheSize)
    {
        if (m_jamming)
        {
            /// 拥塞中, 不追加.
            size_t count = m_pktQueue.dropUntilKeyFrame();
            CLog::warning("evbuffer cache is too much. cache:%d, drop:%d\n", length, count);
        }
        else
        {
            m_jamming = true;
            CLog::warning("HttpMediaCasterConnection(%s) is jamming. buffer size: %d\n", m_peerID.c_str(), length);
            bufferevent_enable(bev, EV_WRITE);
        }
    }
    else if (length > m_maxCacheSize/4)
    {
        if (m_jamming)
        {
            /// 拥塞中, 不追加.
        }
        else
        {
            bufferevent_write(bev, (char*)buf, buf_size);
        }
    }
    else
    {
        if (m_jamming)
        {
            m_jamming = false;
            CLog::warning("HttpMediaCasterConnection(%s) resume from jam: buffer size%d\n", m_peerID.c_str(), length);
        }

        bufferevent_write(bev, (char*)buf, buf_size);
    }

    return buf_size;
}

void HttpMediaCasterConnection::handleRecv(struct bufferevent *bev, HttpMessage& req, std::string& buf)
{
    m_streamState = av::STATE_PLAYING;

    std::string header =
            "HTTP/1.1 200 OK\r\n"
            "Cache-Control: no-cache\r\n"
            "Content-Type: video/MP2T\r\n"
            "Transfer-Encoding: chunked\r\n"
            "\r\n"
            ;
    bufferevent_write(bev, header.c_str(), header.size());

}

bool HttpMediaCasterConnection::canStream()
{
    return m_streamState == av::STATE_PLAYING;
}


} /* namespace av */
